package com.darcytech.provider;

import com.alibaba.edas.acm.ConfigService;
import org.apache.kafka.common.config.ConfigData;
import org.apache.kafka.common.config.ConfigException;
import org.apache.kafka.common.config.provider.ConfigProvider;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.*;

public class AcmConfigProvider implements ConfigProvider {

    private static final Logger LOGGER = LoggerFactory.getLogger(AcmConfigProvider.class);

    public AcmConfigProvider() {

    }

    @Override
    public ConfigData get(String groupAndDataIdStr) {

        Map<String, String> data = new HashMap();
        Properties acmProperties = null;
        if (groupAndDataIdStr != null && !groupAndDataIdStr.isEmpty()) {
            try {
                String[] groupAndDataId = groupAndDataIdStr.split("\\|");
                acmProperties = ConfigService.getConfig2Properties(groupAndDataId[1], groupAndDataId[0], 6000);
                try {
                    Enumeration keys = acmProperties.keys();

                    while (keys.hasMoreElements()) {
                        String key = keys.nextElement().toString();
                        String value = acmProperties.getProperty(key);
                        if (value != null) {
                            data.put(key, value);
                        }
                    }
                    return new ConfigData(data);
                } catch (Throwable e) {
                    throw e;
                }
            } catch (com.alibaba.edas.acm.exception.ConfigException e) {
                throw new ConfigException("Could not read properties from acm " + groupAndDataIdStr);
            }
        } else {
            return new ConfigData(data);
        }
    }

    @Override
    public ConfigData get(String groupAndDataIdStr, Set<String> keys) {
        Map<String, String> data = new HashMap();
        Properties acmProperties = null;
        if (groupAndDataIdStr != null && !groupAndDataIdStr.isEmpty()) {
            try {
                String[] groupAndDataId = groupAndDataIdStr.split("\\|");
                acmProperties = ConfigService.getConfig2Properties(groupAndDataId[1], groupAndDataId[0], 6000);
                try {
                    Iterator keysIterator = keys.iterator();

                    while (keysIterator.hasNext()) {
                        String key = (String) keysIterator.next();
                        String value = acmProperties.getProperty(key);
                        if (value != null) {
                            data.put(key, value);
                        }
                    }
                    return new ConfigData(data);
                } catch (Throwable e) {
                    throw e;
                }
            } catch (com.alibaba.edas.acm.exception.ConfigException e) {
                throw new ConfigException("Could not read properties from acm " + groupAndDataIdStr);
            }
        } else {
            return new ConfigData(data);
        }
    }

    @Override
    public void close() throws IOException {

    }

    @Override
    public void configure(Map<String, ?> map) {
        Properties properties = new Properties();
        LOGGER.info("properties: {}", properties);
        properties.put("endpoint", map.get("endpoint"));
        properties.put("namespace", map.get("namespace"));
        properties.put("accessKey", map.get("accessKey"));
        properties.put("secretKey", map.get("secretKey"));
        ConfigService.init(properties);
    }

}
